package tcp;

import config.ServerConfig;
import database.impl.MulIDB;
import database.inter.IDB;
import database.singleDB.Router;
import handler.inter.Handler;
import jodis.conn.impl.EmptyMultiBulkReply;
import jodis.conn.impl.JoConnection;
import jodis.conn.impl.MultiBulkReply;
import jodis.conn.impl.NullReply;
import jodis.conn.inter.IReply;
import parser.Parser;
import parser.ReadState;
import parser.ReadStateNIO;
import rdb.RDBParser;
import rdb.model.BaseObject;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

public class Server {

    private ExecutorService executorService = Executors.newScheduledThreadPool(10);
    private static ServerConfig serverConfig;

    private Selector selector;

    private IDB db;

    private Logger logger = Logger.getLogger("Server");

    private BufferedOutputStream out;
    private Parser parser;

    public static ServerConfig getServerConfig() {
        return serverConfig;
    }

    public void listenAndServerWithNIO() {
        try {
            serverConfig = ServerConfig.getDefaultConfig();
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(serverConfig.port));
            serverSocketChannel.configureBlocking(false);

            out = new BufferedOutputStream(new FileOutputStream("redis_benchmark.bin"));

            parser = new Parser();

            Router.makeCommandTable();

            MulIDB dbImpl = new MulIDB();
            // 如果开启 aof，则尝试读取 aof 文件中的内容
            dbImpl.loadAof();
            db = dbImpl;

            // 如果没有开启 aof，则尝试读取 rdb 文件
            if (!serverConfig.appendOnly && new File(serverConfig.rdbFileName).exists()) {
                RDBParser parser = new RDBParser(serverConfig.rdbFileName);
                System.out.println("read rdb");
                List<BaseObject> dbObjs = parser.parse();
                if (dbObjs != null) {
                    JoConnection connection = JoConnection.fakeConn(0);
                    for (BaseObject baseObject : dbObjs) {
                        connection.selectDB(baseObject.getDBIndex());
                        db.exec(connection, baseObject.getCmdLines());
                        if (baseObject.getExpiration() != 0) {
                            db.exec(connection, new String[]{"pexpireat", baseObject.getKey(), String.valueOf(baseObject.getExpiration())});
                        }
                    }
                }
            }

            // 如果开启 aof，则打开 aof 写入线程
            if (serverConfig.appendOnly) {
                dbImpl.handleAof();
            }

            selector = Selector.open();

            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            listenFromNIO();
        }  catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void listenFromNIO() throws IOException {
        while (true) {
            Set<SelectionKey> keys = selector.keys();
            while (selector.select() > 0) {
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (selectionKey.isAcceptable()) {
                        // 收到一个客户端连接
                        acceptHandler(selectionKey);
                    } else if (selectionKey.isReadable()) {
//                        readHandler(selectionKey);
                        readHandler2(selectionKey);
                    }
                }
            }
        }
    }

    private void acceptHandler(SelectionKey selectionKey) throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
        SocketChannel accept = serverSocketChannel.accept();
        accept.configureBlocking(false);
        byte[] buf = new byte[4096];
        ByteBuffer byteBuffer = ByteBuffer.wrap(buf);
        accept.register(selector, SelectionKey.OP_READ, byteBuffer);
    }

    private byte[] bufferToBytesArray(ByteBuffer byteBuffer) {
        byte[] ret = new byte[byteBuffer.limit()];
        int size = byteBuffer.limit();
        for (int i=0;i<size;i++) {
            ret[i] = byteBuffer.get();
        }
        return ret;
    }

    private void readHandler2(SelectionKey key) {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        ReadStateNIO state = ReadStateNIO.getDefaultState();
        byte[] msg = null;
        parser.setDb(db);
        parser.setClientConn(new JoConnection(client));

        try {
            while (true) {
                msg = Parser.readLine(client, buffer, state);
                if (msg != null) {
                    parser.parseWithNIO(msg, state);
                } else if (state.reachEOF) {
                    client.close();
                    break;
                } else {
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void readHandler(SelectionKey key) {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        int read;
        try {
            while (true) {
                read = client.read(buffer);
                if (read > 0) {
                    buffer.flip();
                    handleWithNIO(bufferToBytesArray(buffer), client);
                    buffer.clear();
                } else if (read == 0) {
                    System.out.println("read == 0");
                    break;
                } else {
                    System.out.println("client close");
                    client.close();
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void handleWithNIO(byte[] bs, SocketChannel client) throws IOException {
        ReadState state = ReadState.getDefaultInstance();
        out.write(bs);
        out.flush();
        Parser.parse(bs, state);
        JoConnection clientConn = new JoConnection(client);
        if (state.msgType == Parser.ARRAY) {
            if (state.resultArr[0].equalsIgnoreCase("command")) {
                clientConn.writeWithNIO(EmptyMultiBulkReply.makeEmptyMultiBulkReply().toBytes());
                return;
            }
            // 交给对应的 db 去处理
            IReply reply = db.exec(clientConn, state.resultArr);
            clientConn.writeWithNIO(reply.toBytes());
        } else if (state.msgType != Parser.NONE) {
            logger.info(state.resultStr);
        } else {
            logger.info(String.valueOf(state.msgType));
        }
    }


    public void listenAndServerWithSignal() {
        try {
            // 暂时先用默认配置
            serverConfig = ServerConfig.getDefaultConfig();
            ServerSocket serverSocket = new ServerSocket(serverConfig.port);
            Router.makeCommandTable();
            listenAndServer(serverSocket, jodis.Server.makeHandler());
        } catch (IOException e) {
             e.printStackTrace();
        }
    }

    public void listenAndServer(ServerSocket serverSocket, Handler handler) {
        try {
            while (true) {
                Socket socket = serverSocket.accept();
                // 每收到一个客户端连接，就开一个线程来处理
                ConnTask task = new ConnTask(socket, handler);
//                Thread thread = new Thread(task);
                executorService.execute(task);
            }

        } catch (IOException e) {

        }

    }

    public static void main(String[] args) {
        Server tcpServer = new Server();
//        tcpServer.listenAndServerWithSignal();
        tcpServer.listenAndServerWithNIO();

    }


}


